#ifndef K_MULTIPLEXOR_H
#define K_MULTIPLEXOR_H
#pragma once

#include "KAutoPtr.h"
#include "KDef.h"
#include <functional>

class KIAddress;
class KPackage;
class KSocketIocp;
class KStorage;
class KTaskThread;

class KMultiplexor : KNoCopy {
public:
    typedef std::function<bool (KSocket *)> Condition;
public:
    explicit KMultiplexor(size_t numOfWorkers);
    virtual ~KMultiplexor() = 0;
public:
    bool isRunning() const                      { return !m_stopped; }
    bool start()                                { return startImp(); }
    bool stop(DWORD ms = KDef::DefaultTimeOut)  { return stopImp(ms); }
protected:
    size_t chunkSize() const                    { return m_chunkSize; }
    void setChunkSize(size_t size)              { m_chunkSize = size ? size : KDef::DefaultChunkSize; }
    size_t maxPoolSize() const                  { return m_pool.maxSize(); }
    void setMaxPoolSize(size_t size = 0)        { m_pool.setMaxSize(size); }
    size_t poolSize() const                     { return m_pool.size(); }
protected:
    // TCP
    KSocket *addAcceptor(const KIAddress &addrLocal,
        KKey *key = NULL, int sendBufSize = -1, bool reuseAddrLocal = false, size_t extra = 0);
    KSocket *addConnector(const KIAddress &addrLocal, const KIAddress &addrRemote,
        KKey *key = NULL, int sendBufSize = -1, bool reuseAddrLocal = false);
    // UDP
    KSocket *addMessenger(const KIAddress &addrLocal,
        KKey *key = NULL, int sendBufSize = -1, bool reuseAddrLocal = false);
protected:
    // For both TCP and UDP
    bool cancel(KSocket *sock, KDef::BasicRequest *request = NULL);
    bool postMessage(KSocket *sock, KKey *key = NULL);
    bool receive(KSocket *sock, size_t length = 0, KKey *key = NULL);
    // For TCP only
    bool disconnect(KSocket *sock);
    bool send(KSocket *sock, KStorage *storage, size_t length = 0);
    bool send(KSocket *sock, const void *data, size_t length, KKey *key = NULL);
    bool transmit(KSocket *sock, HANDLE hFile, size_t length = 0, KKey *key = NULL, size_t blockLen = 0, KPackage *head = NULL, KPackage *tail = NULL);
    // For TCP server only
    size_t disconnectIf(const Condition &con);
    size_t sendIf(const void *data, size_t length, const Condition &con);
    // For UDP only
    bool sendTo(KSocket *sock, const KIAddress &addrRemote, const void *data, size_t length, KKey *key = NULL);
protected:
    // For lifetime control
    void ensureAlive(KSocket *sock, bool always = true);
    KDef::Life lifeBegin(KSocket *sock);
    void lifeEnd(KSocket *sock, KDef::Life life);
protected:
    // For TCP only
    virtual void onAccepted(KSocket *sock, KKey *key) {}
    virtual void onConnected(KSocket *sock) {}
    virtual void onDisconnected(KSocket *sock) {}
    virtual void onTransmitted(KSocket *sock, KKey *&key, HANDLE file, size_t length) {}
    // For UDP only
    virtual void onCreated(KSocket *sock) {}
    // For both TCP and UDP
    virtual void onClosing(KSocket *sock, bool stopped) {}
    virtual void onError(KSocket *sock, KDef::SockError type, DWORD error) {}
    virtual void onMessage(KSocket *sock, KKey *&key) {}
    virtual void onReceived(KSocket *sock, KKey *&key, void *&data, size_t length, bool more) {}
    virtual void onSent(KSocket *sock, KKey *&key, void *&data, size_t length) {}
    virtual void onStopping() {}
    // Implementation
    virtual bool startImp();
    virtual bool stopImp(DWORD ms);
private:
    void _gc();
    void _worker();
    bool accept(KSocket *sock, KDef::AcceptRequest *request = NULL);
    bool connect(KSocket *sock, const KIAddress &addr);
    void deleteLater(KSocket *sock);
    bool sendAgain(KSocket *sock, KDef::BasicRequest *request, WSABUF *buffer);
    KList::NodePtr validate(KDef::SocketContext *node);
private:
    volatile bool               m_stopped;
    KAutoEvent                  m_gcStop;
    KAutoPtr<KSocketIocp>       m_iocp;
    KAutoPtr<KTaskThread>       m_gc;
    KAutoPtr<KTaskThread, true> m_workers;
    KList                       m_context;
    KList                       m_pool;
    size_t                      m_chunkSize;
    size_t                      m_nThreads;
    friend class                KDef::RequestInfo;
};

#endif // K_MULTIPLEXOR_H